package com.java1234.rocketmq;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Description: 顺序消息
 * <p>
 * RocketMQTemplate 给我们提供了SendOrderly方法(有多个重载)，来实现发送顺序消息；包括以下：
 * syncSendOrderly，发送同步顺序消息；
 * asyncSendOrderly，发送异步顺序消息；
 * sendOneWayOrderly，发送单向顺序消息；
 * <p>
 * <p>
 * <p>
 * 以及消费端接收的时候，默认是并发多线程去接收消息。RocketMQMessageListener有个属性consumeMode，
 * 默认是ConsumeMode.CONCURRENTLY ，我们要改成ConsumeMode.ORDERLY，单线程顺序接收消息；
 */
@Component("orderlyProducerService")
public class OrderlyProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /*** 发送同步顺序消息 */
    public void sendOrderlyMessage() {
        // hashKey用来计算决定消息发送到哪个消息队列 MessageQueue 一般是订单ID，产品ID等
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456231,创 建", "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456231,支 付", "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456231,完 成", "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,创 建", "98456232");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,支 付", "98456232");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,完 成", "98456232");
    }

    /*** 发送延迟消息 */
    // messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    // 如果过了延时消息时间后 还没消费 后面消费者还是可以消费的，当做普通消息来消费
    public void sendDelayMessage() {
        rocketMQTemplate.syncSend("java1234-rocketmq", MessageBuilder.withPayload("rocketmq延迟1秒消息").build(), 3000, 1);
        rocketMQTemplate.syncSend("java1234-rocketmq", MessageBuilder.withPayload("rocketmq延迟5秒消息").build(), 3000, 2);
        rocketMQTemplate.syncSend("java1234-rocketmq", MessageBuilder.withPayload("rocketmq延迟10秒消息").build(), 3000, 3);
    }

    /**
     * 1. 应用模块遇到要发送事务消息的场景时，先发送prepare消息给MQ。 2. prepare消息发送成功后，应用模块执行数据库事务（本地事务）。
     * 3. 根据数据库事务执行的结果，再返回Commit或Rollback给MQ。 4. 如果是Commit，MQ把消息下发给Consumer端，如果是Rollback，直接删掉prepare消息。
     * 5. 第3步的执行结果如果没响应，或是超时的，启动定时任务回查事务状态（最多重试15次，超过了
     * 默认丢弃此消息），处理结果同第4步。
     * 6. MQ消费的成功机制由MQ自己保证。
     */

    /*** 发送事务消息 */
    public void sendTransactionMessage() {
        // 构造消息
        Message msg = MessageBuilder.withPayload("rocketmq事务消息-01").build();
        rocketMQTemplate.sendMessageInTransaction("Tx_Charge_Group", "java1234-transaction-rocketmq", msg, null);
    }

    /**
     * executeLocalTransaction 方法，当我们处理完业务后，可以根据业务处理情况，返回事务执行状态，
     * 有 bollback , commit or unknown 三种：分别是回滚事务，提交事务和未知；根据事务消息执行流程，
     *
     *  1、如果返回bollback，则直接丢弃消息；
     *  2、如果是返回commit，则消费消息；
     *  3、如果是unknow，则继续等待，然后调用checkLocalTransaction方法，最多重试15次，超过了默认丢弃此消息；
     */
    @RocketMQTransactionListener(txProducerGroup = "Tx_Charge_Group") // 这里的txProducerGroup的值要与发送半消息时保持一致
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            // ... local transaction process, return bollback, commit or unknown
            System.out.println("executeLocalTransaction");
            return RocketMQLocalTransactionState.UNKNOWN; // 因为我返回unknown 所以会不断调用checkLocalTransaction(最多15次)
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // ... check transaction status and return bollback, commit or unknown
            System.out.println("checkLocalTransaction");
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }


}
